package com.wulaobo.state;

import com.wulaobo.constant.Constants;
import com.wulaobo.functions.WaterSensorMapFunction;
import com.wulaobo.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Objects;

/**
 * 案例需求：检测每种传感器的水位值，如果连续的两个水位值超过10，就输出报警
 */
public class KeyedValueStateDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream(Constants.IP, 7777)
                .map(new WaterSensorMapFunction());
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                                .withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
//                );

        sensorDS.keyBy(e -> e.getId()).process(new KeyedProcessFunction<String, WaterSensor, String>() {

            // TODO 1.定义状态
            ValueState<Integer> lastVcState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);

                // TODO 2.在open方法中，初始化状态
                // 状态描述器两个参数：第一个参数，起个名字，不重复；第二个参数，存储的类型
                lastVcState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("lastVcState", Types.INT));
            }

            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                int lastVc = Objects.isNull(lastVcState.value()) ? 0 : lastVcState.value();
                if(Math.abs(value.getVc()-lastVc)>10){
                    out.collect("key:"+value.getId()+"最近两条水位线的差值大于10 lastVc："+lastVc+",currentVC:"+value.getVc());
                }
                lastVcState.clear();
                out.collect("key:"+value.getId()+" clear之后的lastVcState的值为："+lastVcState.value());
                lastVcState.update(value.getVc());
                out.collect("key:"+value.getId()+" lastVcState的值为："+lastVcState.value());

            }
        }).print();

        env.execute();

    }

}

